package com.gandong8.es.tool.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.message.BasicHeader;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.bulk.*;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.*;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.ArrayList;
import java.util.function.BiConsumer;

@Configuration
@Slf4j
public class EsConf {

    /**
     * 集群地址，多个用,隔开
     */
    @Value("${es.cluster.nodes:localhost:9200}")
    private  String clusterNodes;
    /**
     * 使用的协议
     */
    @Value("${es.schema:http}")
    private  String schema;

    private  int connectTimeOut = 1000;
    private  int socketTimeOut = 3000;
    private  int connectionRequestTimeOut = 500;
    private  int maxConnectNum = 100;
    private  int maxConnectPerRoute = 100;


    @Bean(name = "restHighLevelClient")
    public RestHighLevelClient client(){
        ArrayList<HttpHost> hostList = new ArrayList<>();
        String[] arr = clusterNodes.split(",");
        for (String node : arr) {
            String[] data = node.split(":");
            hostList.add(new HttpHost(data[0], Integer.parseInt(data[1]), schema));
        }
        HttpHost[] httpHosts = hostList.toArray(new HttpHost[hostList.size()]);
        RestClientBuilder builder = RestClient.builder(httpHosts);
        //节点发生故障时收到通知的监听器
        builder.setFailureListener(new RestClient.FailureListener() {
            @Override
            public void onFailure(Node node) {
                log.error("node_failed:{}",node);
            }
        });
        //header配置
        Header[] defaultHeaders = new Header[]{
                new BasicHeader("header", "test")
        };
        builder.setDefaultHeaders(defaultHeaders);
        //连接延时配置
        builder.setRequestConfigCallback(requestConfigBuilder -> {
            requestConfigBuilder.setConnectTimeout(connectTimeOut);
            requestConfigBuilder.setSocketTimeout(socketTimeOut);
            requestConfigBuilder.setConnectionRequestTimeout(connectionRequestTimeOut);
            return requestConfigBuilder;
        });
        //身份认证
        final CredentialsProvider credentialsProvider =
                new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials("user", "password"));
        //连接数配置
        builder.setHttpClientConfigCallback(httpClientBuilder -> {
            httpClientBuilder.setMaxConnTotal(maxConnectNum);
            httpClientBuilder.setMaxConnPerRoute(maxConnectPerRoute);
            //httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            //httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(10).build());
            return httpClientBuilder;
        });
        client =  new RestHighLevelClient(builder);
        return client;
    }

    private RestHighLevelClient client;

    @Bean(name = "bulkProcessor")
    public BulkProcessor bulkProcessor(){
        BiConsumer<BulkRequest, ActionListener<BulkResponse>> bulkConsumer =
                (request, bulkListener) -> client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener);
        return BulkProcessor.builder(bulkConsumer, new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                log.warn("numberOfActions:{}", request.numberOfActions());
            }
            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                for (BulkItemResponse itemResponse : response) {
                    if (itemResponse.getFailure() != null) {
                        if (itemResponse.getOpType() == DocWriteRequest.OpType.INDEX
                                    || itemResponse.getOpType() == DocWriteRequest.OpType.CREATE) {
                                IndexResponse indexResponse =  itemResponse.getResponse();

                        } else if (itemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) {
                            UpdateResponse updateResponse = itemResponse.getResponse();

                        } else if (itemResponse.getOpType() == DocWriteRequest.OpType.DELETE) {
                            DeleteResponse deleteResponse =  itemResponse.getResponse();
                        }
                        log.error("bulk_failed:{},{},{}", itemResponse.getIndex(),itemResponse.getId(),itemResponse.getFailureMessage());
                    }
                }
            }
            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                for (Object o : request.requests()) {
                    String value = o.toString();
                    log.error("bulk_err:{}", value);
                }
                log.error("BulkProcessor_failed:", failure);
            }
        }).setBulkActions(1000)
                .setBulkSize(new ByteSizeValue(1, ByteSizeUnit.MB))
                .setFlushInterval(TimeValue.timeValueSeconds(5))
                .setConcurrentRequests(1)
                .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3))
                .build();
    }

}
